﻿using System;
using System.Buffers;
using System.IO.Pipelines;
using System.Threading.Tasks;
using Deer.Time;
using Deer.Util;
using Letter.IO.Binary;
using Letter.IO.Channels;
using Telex.Timer;

namespace Telex;

delegate void PipelineMessageDelegate(ref SocketPak pak);

sealed class RpcPipeline : IComparable<RpcPipeline>, IAsyncDisposable
{
    private const int FIXED_PAK_SIZE = sizeof(byte) + sizeof(uint);

    public RpcPipeline(ARcpConnection connection, bool heartbeatSponsor)
    {
        this.connection = connection;
        this.sndChannel = ChannelProvider.CreateDataChannel<SocketPak>(OnMessageChannelNotify);

        this.connection.OnSocketMessageDelegate += RcvSocketPak;
        
        this.SndChannelReader.AsyncRcv();

        if (heartbeatSponsor)
        {
            this.nextTimestamp = TimeUtil.GetTimestamp(DateTime.UtcNow.AddSeconds(15));
            SingleTimer.instance.AddListener(SponsorHeartbeat);
        }
    }

    public string Id
    {
        get { return this.connection.Id; }
    }
    
    public RpcHost LocalHost
    {
        get { return this.connection.LocalHost; }
    }

    public RpcHost RemoteHost
    {
        get { return this.connection.RemoteHost; }
    }

    private IChannelDataReader<SocketPak> SndChannelReader
    {
        get { return this.sndChannel.Reader.ToChannelDataReader(); }
    }

    private IChannelDataWriter<SocketPak> SndChannelWriter
    {
        get { return this.sndChannel.Writer.ToChannelDataWriter(); }
    }


    private ARcpConnection connection;
    private IChannel<SocketPak> sndChannel;

    private PipelineMessageDelegate func;
    private long nextTimestamp;

    public int Pressure
    {
        get
        {
            if (this.SndChannelReader.Active)
            {
                return (int)this.SndChannelReader.Count;
            }

            return int.MaxValue;
        }
    }
    
    public int CompareTo(RpcPipeline other)
    {
        if (this.Pressure == other.Pressure)
        {
            return 0;
        }

        if (this.Pressure < other.Pressure)
        {
            return 1;
        }

        return -1;
    }

    public void AddPipelineMessageListener(PipelineMessageDelegate func)
    {
        if (func == null)
        {
            return;
        }

        this.func += func;
    }

    public void DelPipelineMessageListener(PipelineMessageDelegate func)
    {
        if (func == null)
        {
            return;
        }

        this.func -= func;
    }

    private void SponsorHeartbeat(long timestamp)
    {
        if (timestamp >= this.nextTimestamp)
        {
            var pak = new SocketPak()
            {
                command = Command.HEARTBEAT,
                hash = 0,
                buf = Array.Empty<byte>()
            };

            this.TryWriteSndChannel(ref pak);
            
            this.nextTimestamp = TimeUtil.GetTimestamp(DateTime.UtcNow.AddSeconds(15));
        }
    }

    public bool TryWriteSndChannel(ref SocketPak pak)
    {
        var b = this.SndChannelWriter.TryWrite(pak);
        this.SndChannelWriter.Flush();

        return b;
    }

    private async void OnMessageChannelNotify(object state)
    {
        while (this.SndChannelReader.TryRead(out var pak))
        {
            var result = await SndSocketPak(ref pak);
            if (result.IsCanceled || result.IsCompleted)
            {
            }
        }

        this.SndChannelReader.AsyncRcv();
    }
    
    private void RcvSocketPak(byte command, ref BytesReader reader)
    {
        var hash = reader.ReadUInt32BE();
        var buf = reader.ReadBuffer((int)reader.ReadableLength);
        
        int size = (int)buf.Length;
        var sourceArr = ArrayPool<byte>.Shared.Rent(size);
        var segment = sourceArr.AsSegment(0, size);
        buf.CopyTo(segment);

        var pak = new SocketPak()
        {
            command = command,
            hash = hash,
            buf = segment
        };
        
        this.func?.Invoke(ref pak);
    }
    
    private ValueTask<FlushResult> SndSocketPak(ref SocketPak pack)
    {
        var output = this.connection.Pipelines.Output;
        var writer = output.ToBytesWriter();
        writer.WriteInt32BE(FIXED_PAK_SIZE + pack.buf.Count);
        writer.WriteUInt8(pack.command);
        writer.WriteUInt32BE(pack.hash);
        writer.WriteSpan(pack.buf.AsSpan());

        return output.FlushAsync();
    }
    
    public ValueTask DisposeAsync()
    {
        this.SndChannelWriter.Complete();
        return ValueTaskUtil.CompletedTask;
    }
}